# 从本地文件系统中加载数据创建RDD

from pyspark import SparkConf, SparkContext
conf = SparkConf().setMaster("local").setAppName("My App") #用单机模式启动
sc = SparkContext(conf = conf)

'''
# 从本地系统中加载数据
lines = sc.textFile("file:///home/hadoop/MyTmp/sparkwordcount.txt")

lines.foreach(print) #循环遍历文件的每一行
'''


'''
# 从分布式文件系统 HDFS 中加载数据
lines = sc.textFile("hdfs://localhost:9000/user/hadoop/sparkwordspace/sparkwordcount.txt")
lines1 = sc.textFile("/user/hadoop/sparkwordspace/sparkwordcount.txt")
lines2 = sc.textFile("sparkwordcount.txt")

lines.foreach(print) #循环遍历文件的每一行
'''

'''
# 通过并行集合(列表)创建 RDD
# 可以调用 SparkContext 的 parallelize 方法,在 Driver 中一个已经存在的集合 (列表)上创建。
array = [1,2,3,4,5]
rdd = sc.parallelize(array)
rdd.foreach(print)

'''





'''
# 1. 转换操作
# filter(func) :筛选出满足函数 func 的元素,并返回一个新的数据集
lines = sc.textFile("file:///home/hadoop/MyTmp/sparkwordcount.txt")
linesWith_woaini = lines.filter(lambda line: "woaini" in line)
linesWith_woaini.foreach(print)
'''



'''
# map(func) 操作将每个元素传递到函数 func 中,并将结果返回为一个新的数据集
data = [1,2,3,4,5]
rdd1 = sc.parallelize(data)
rdd1.foreach(print)
print("*"*50)
rdd2 = rdd1.map(lambda x:x+10)#rdd1中的每个元素进行加10操作
rdd2.foreach(print)
'''



'''
# map(func)
# # 另外一个实例
lines = sc.textFile("file:///home/hadoop/MyTmp/sparkwordcount.txt")
words = lines.map( lambda line:line.split(" ")) #对每行进行空格拆分，得到列表类型
words.foreach(print)
'''



'''
# flatMap(func)
# flatMap(func) 与map() 相似，但每个输入元素都可以映射到0或多个输出结果
lines = sc.textFile("file:///home/hadoop/MyTmp/sparkwordcount.txt")
word = lines.flatMap(lambda line :line.split(" "))
word.foreach(print)

'''



'''
# groupByKey()
# groupByKey() 应用于 (K,V) 键值对的数据集时,返回一个新的 (K, Iterable) 形式的数据集
words = sc.parallelize([("Hadoop",1),("is",1),("good",1), ("Spark",1),("is",1),("fast",1),("Spark",1),("is",1),("better",1)])
words1 = words.groupByKey()
words1.foreach(print)
'''

'''
# reduceByKey(func)
# reduceByKey(func) 应用于 (K,V) 键值对的数据集时,返回一个新的 (K, V)形式的数据集,其中的每个值是将每个 key 传递到函数 func 中进行聚合后得到的结果
words = sc.parallelize([("Hadoop",1),("is",1),("good",1), ("Spark",1),("is",1),("fast",1),("Spark",1),("is",1),("better",1)])
words1 = words.reduceByKey(lambda a,b:a+b)
words1.foreach(print)
'''


# 行动操作
'''
# count():返回数据集中的元素个数
array = [1,2,3,4,5]
rdd = sc.parallelize(array)
wordsCount = rdd.count()
print(wordsCount)

'''


'''
# first()返回数据集中的第一个元素
array = [1,2,3,4,5]
rdd = sc.parallelize(array)
firstWord = rdd.first()
print('firstWord:',firstWord)
'''


'''
# collect() 以数组的形式返回数据集中的所有元素
array = [1,2,3,4,5]
rdd = sc.parallelize(array)
rdd.collect()
rdd.foreach(lambda elem:print(elem))
'''


'''
#take(n) 以数组的形式返回数据集中的前 n 个元素
array = [1,2,3,4,5]
rdd = sc.parallelize(array)
takeResult = rdd.take(3)
print(takeResult)
'''


'''
# reduce(func) 通过函数 func (输入两个参数并返回一个值)聚合数据集中的元素
array = [1,2,3,4,5]
rdd = sc.parallelize(array)
print(rdd.reduce(lambda a,b:a+b))
'''

'''
# foreach(func) 将数据集中的每个元素传递到函数 func 中运行
array = [1,2,3,4,5]
rdd = sc.parallelize(array)
rdd.foreach(lambda elem:print(elem))
'''


'''
# 惰性机制
# 所谓的“惰性机制”是指,整个转换过程只是记录了转换的轨迹,并不会发
# 生真正的计算,只有遇到行动操作时,才会触发“从头到尾”的真正的计算
# 这里给出一段简单的语句来解释 Spark 的惰性机制
lines  = sc.textFile("file:///home/hadoop/MyTmp/sparkwordcount.txt")
lineLengths = lines.map(lambda s:len(s))
totalLength = lineLengths.reduce(lambda a,b:a+b)
print(totalLength)
'''

'''
# 持久化
# 在 Spark 中, RDD 采用惰性求值的机制,每次遇到行动操作,都会从头开始 执行计算。每次调用行动操作,都会触发一次从头开始的计算。
# 这对于迭代计算而言,代价是很大的,迭代计算经常需要多次重复使用同一组数据
array = ["hadoop","spark","Hive"]
rdd = sc.parallelize(array)
rdd.cache()# 会调用 persist(MEMORY_ONLY) ,但是,语句执行到这里,并不会缓存 rdd ,因为这时 rdd 还没有被计算生成
print(rdd.count()) # 第一次行动操作,触发一次真正从头到尾的计算,这时上面的 rdd.cache() 才会被执行,把这个 rdd 放到缓存中
print(','.join(rdd.collect()))  # 第二次行动操作,不需要触发从头到尾的计算,只需要重复使用上面缓存中的 rdd
rdd.unpersist() # 手动地把持久化的 RDD 从缓存中移除
'''


'''
# 创建 RDD 时手动指定分区个数
# 在调用 textFile() 和 parallelize() 方法的时候手动指定分区个数即可,语法格式如下:
# sc.textFile(path, partitionNum)其中, path 参数用于指定要加载的文件的地址, partitionNum 参数用于指定分区个数。
array = [1,2,3,4]
rdd = sc.parallelize(array,2) # 设置两个分区
'''

# 设置分区的个数
# ( 2 )使用 reparititon 方法重新设置分区个数
# 通过转换操作得到新 RDD 时,直接调用 repartition 方法即可。例如:
array = [1,2,3,4,5]
data = sc.parallelize(array,2)
print(len(data.glom().collect())) # 显示 data 这个 RDD 的分区数量
rdd = data.repartition(1) # 对 data 这个 RDD 进行重新分区
print(len(rdd.glom().collect())) # 显示 rdd 这个 RDD 的分区数量